package com.gitee.fsmxm.thinking.in.concurrent.basic;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 阻塞队列
 * 用 ReentrantLock Condition 实现
 */
public class _05_ReentrantLock_BlockedQueue {

    public static class BlockedQueue<T> {
        final ReentrantLock lock;
        final Condition notFull;
        final Condition notEmpty;

        final Object[] items;

        int count;
        int takeIndex; // 出队 索引 (循环索引)
        int putIndex;  // 入队 索引 (循环索引)

        public BlockedQueue(int size) {
            items = new Object[size];
            lock = new ReentrantLock();
            notFull = lock.newCondition();
            notEmpty = lock.newCondition();
        }

        public void put(T t) throws InterruptedException {
            checkNotNull(t);
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
                while (count == items.length) {
                    notFull.await();
                }
                enqueue(t);
            } finally {
                lock.unlock();
            }
        }

        private void checkNotNull(T t) {
            if (t == null) {
                throw new NullPointerException();
            }
        }

        public T take() throws InterruptedException {
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
                while (count == 0) {
                    notEmpty.await();
                }
                return dequeue();
            } finally {
                lock.unlock();
            }
        }

        private void enqueue(T t) {
            final Object[] items = this.items;
            items[putIndex] = t;
            if (++putIndex == items.length) {
                putIndex = 0;
            }
            count++;
            notEmpty.signal();
        }

        private T dequeue() {
            final Object[] items = this.items;
            @SuppressWarnings("unchecked")
            T t = (T) items[takeIndex];
            items[takeIndex] = null;
            if (++takeIndex == items.length) {
                takeIndex = 0;
            }
            count--;
            notFull.signal();
            return t;
        }
    }

    public static void main(String[] args) throws InterruptedException {
        final BlockedQueue<Integer> queue = new BlockedQueue<>(1);
        Thread t1 = new Thread(() -> {
            try {
                queue.put(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, "t1");
        Thread t2 = new Thread(() -> {
            try {
                queue.put(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, "t2");
        Thread t3 = new Thread(() -> {
            try {
                System.out.println("take: " + queue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, "t3");

        t1.start();
        t2.start();
        t3.start();

        t1.join();
        t2.join();
        t3.join();

        System.out.println(queue.take());
    }

}
